package com.smx.sync.service;

import com.smx.sync.entity.SyncTask;
import com.smx.sync.entity.SyncTaskDaily;
import com.smx.sync.entity.SyncTaskLog;
import com.smx.sync.mapper.SyncConfigMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.*;

/**
 * SyncTaskExecutor
 *
 * @author 廖功旺
 * @version 1.0
 * @since 2025/6/23 17:29
 */
@Slf4j
@Service
public class SyncTaskExecutor {

    @Autowired
    private SyncConfigMapper syncConfigMapper;
    // 时间格式 yyyy-MM-dd HH:mm:ss
    private static final SimpleDateFormat SDF_DATETIME = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    // 时间格式 yyyy-MM-dd
    private static final SimpleDateFormat SDF_DATE = new SimpleDateFormat("yyyy-MM-dd");

    /**
     * 执行数据同步任务，包括从源数据库读取数据、判断新增或更新，并将数据同步到目标数据库。
     * 该方法为事务方法，确保数据同步与状态更新具备原子性。
     *
     * @param task 同步任务配置对象，包含源库、目标库信息及同步规则
     * @throws RuntimeException 如果同步过程中发生异常，将抛出运行时异常并触发事务回滚
     */
    @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRED)
    public void executeSync(SyncTask task) {
        Integer taskId = task.getId();
        String recordDate = SDF_DATE.format(new Date());
        long start = System.currentTimeMillis();
        // log
        SyncTaskLog syncTaskLog = new SyncTaskLog();
        syncTaskLog.setTaskId(taskId);
        syncTaskLog.setStartTime(SDF_DATETIME.format(new Date()));
        syncTaskLog.setRecordDate(recordDate);

        // daily
        SyncTaskDaily syncTaskDaily = new SyncTaskDaily();
        syncTaskDaily.setTaskId(taskId);
        syncTaskDaily.setRecordDate(recordDate);
        saveSyncTaskDaily(taskId, recordDate, syncTaskDaily);
        Integer syncCount = syncTaskDaily.getSyncCount();

        // 执行数据同步
        try {
            log.info("任务ID：{} 开始同步@{}", taskId, SDF_DATETIME.format(new Date()));
            syncData(task, syncTaskLog, syncTaskDaily);
            log.info("任务ID：{} 同步完成@{}", taskId, SDF_DATETIME.format(new Date()));

            // 新增同步任务日志 成功
            long end = System.currentTimeMillis();
            syncTaskLog.setEndTime(SDF_DATETIME.format(new Date()));
            syncTaskLog.setCostTime((int) (end - start));
            syncTaskLog.setStatus(0);
            syncConfigMapper.saveSyncTaskLog(syncTaskLog);

            // 更新当天同步成功次数
            syncTaskDaily.setSuccessCount(syncTaskDaily.getSuccessCount() + 1);
            syncConfigMapper.updateSyncTaskDaily(syncTaskDaily);
        } catch (Exception e) {
            log.error("任务ID：{} 同步失败@{}", taskId, SDF_DATETIME.format(new Date()));

            // 新增同步任务日志 失败
            long end = System.currentTimeMillis();
            syncTaskLog.setEndTime(SDF_DATETIME.format(new Date()));
            syncTaskLog.setCostTime((int) (end - start));
            syncTaskLog.setSyncCount(0);
            syncTaskLog.setStatus(1);
            syncConfigMapper.saveSyncTaskLog(syncTaskLog);

            // 更新当天同步失败次数
            syncTaskDaily.setFailCount(syncTaskDaily.getFailCount() + 1);
            // 回滚当天同步数量
            syncTaskDaily.setSyncCount(syncCount);
            syncConfigMapper.updateSyncTaskDaily(syncTaskDaily);

            throw new RuntimeException("数据同步失败", e);
        }
    }

    // 当天尚未存在日同步任务记录时，则创建
    private void saveSyncTaskDaily(Integer taskId, String recordDate, SyncTaskDaily syncTaskDaily) {
        SyncTaskDaily daily = syncConfigMapper.getSyncTaskDaily(taskId, recordDate);
        if (daily == null) {
            syncTaskDaily.setSuccessCount(0);
            syncTaskDaily.setFailCount(0);
            syncTaskDaily.setSyncCount(0);
            syncConfigMapper.saveSyncTaskDaily(syncTaskDaily);
        } else {
            syncTaskDaily.setSuccessCount(daily.getSuccessCount());
            syncTaskDaily.setFailCount(daily.getFailCount());
            syncTaskDaily.setSyncCount(daily.getSyncCount());
        }
    }

    /**
     * 根据字段映射关系，获取指定 value 对应的 key。
     * 若存在多个相同 value，返回第一个匹配项。
     *
     * @param map   字段映射关系，key 为源字段，value 为目标字段
     * @param value 要查找的目标字段值
     * @return 对应的源字段（key），若未找到则返回 null
     */
    private <K, V> K getKeyByValue(Map<K, V> map, V value) {
        for (Map.Entry<K, V> entry : map.entrySet()) {
            if (value.equals(entry.getValue())) {
                return entry.getKey();
            }
        }
        return null;
    }

    /**
     * 实际执行数据同步逻辑的方法，包括：
     * 1. 获取源数据库连接
     * 2. 查询源数据（全量或增量）
     * 3. 判断记录是新增还是更新
     * 4. 执行批量插入或更新操作
     * 5. 更新同步时间至 sync_time 表
     *
     * @param task 同步任务配置对象
     * @throws Exception 同步过程中的任何异常都会被上层捕获并触发回滚
     */
    public void syncData(SyncTask task, SyncTaskLog syncTaskLog, SyncTaskDaily syncTaskDaily) throws Exception {
        Integer taskId = task.getId();
        Map<String, String> fieldMapping = task.getFieldMappingMap();

        try (Connection sourceConn = DriverManager.getConnection(
                task.getSourceDbUrl(),
                task.getSourceDbUsername(),
                task.getSourceDbPassword());

             Connection targetConn = DriverManager.getConnection(
                     task.getTargetDbUrl(),
                     task.getTargetDbUsername(),
                     task.getTargetDbPassword())) {
            // 获取上次同步时间
            String lastSyncTime = syncConfigMapper.getLastSyncTime(taskId);
            // 记录本次同步使用的最新同步时间
            syncTaskLog.setLastSyncTime(lastSyncTime);
            log.info("任务ID：{} 上次同步时间：{}@{}", taskId, lastSyncTime, SDF_DATETIME.format(new Date()));
            if (!StringUtils.isEmpty(lastSyncTime)) {
                task.setLastSyncTime(SDF_DATETIME.parse(lastSyncTime));
            }

            // 查询源数据（区分首次同步 or 增量同步）
            List<Map<String, Object>> sourceData;
            if (task.getLastSyncTime() == null) {
                sourceData = loadAllSourceData(task, sourceConn);
            } else {
                sourceData = loadNewSourceDataSince(task, sourceConn);
            }
            log.info("任务ID：{} 查询到待同步数据量：{}条", taskId, sourceData.size());

            if (sourceData.isEmpty()) {
                log.info("任务ID：{} 本次没有需要同步的数据@{}", taskId, new Date());
                return;
            }

            // 记录本次同步数量
            syncTaskLog.setSyncCount(sourceData.size());
            // 累计当天同步数量
            syncTaskDaily.setSyncCount(syncTaskDaily.getSyncCount() + sourceData.size());

            // 提取最大时间字段
            String timeField = task.getSourceTimeField();
            Date maxTime = sourceData.stream()
                    .map(record -> {
                        Object timeValue = record.get(timeField);
                        try {
                            if (timeValue instanceof java.time.LocalDateTime) {
                                java.time.LocalDateTime localDateTime = (java.time.LocalDateTime) timeValue;
                                // 将 LocalDateTime 转换为 Date
                                return Date.from(localDateTime.atZone(java.time.ZoneId.systemDefault()).toInstant());
                            } else if (timeValue instanceof java.sql.Timestamp) {
                                // 处理 Timestamp 类型
                                return new Date(((java.sql.Timestamp) timeValue).getTime());
                            } else if (timeValue instanceof String) {
                                // 处理字符串类型
                                return SDF_DATETIME.parse((String) timeValue);
                            } else if (timeValue instanceof Date) {
                                // 已经是 Date 类型
                                return (Date) timeValue;
                            } else {
                                log.warn("不支持的时间字段类型: {}, 值: {}",
                                        timeValue != null ? timeValue.getClass().getName() : "null",
                                        timeValue);
                                return null;
                            }
                        } catch (Exception e) {
                            log.error("时间字段转换失败: {}, 值: {}", e.getMessage(), timeValue);
                            return null;
                        }
                    })
                    .filter(Objects::nonNull)
                    .max(Date::compareTo)
                    .orElse(null);
            log.info("任务ID：{} 最大时间字段：{} 最大时间：{}@{}", taskId, timeField, maxTime,
                    SDF_DATETIME.format(new Date()));

            // 分离新增与更新
            List<Map<String, Object>> insertList = new ArrayList<>();
            List<Map<String, Object>> updateList = new ArrayList<>();

            String idField = task.getSourceIdField();

            for (Map<String, Object> record : sourceData) {
                Object idValue = record.get(idField);
                boolean existsInTarget = checkIfExists(targetConn, task, idValue);
                if (existsInTarget) {
                    updateList.add(record);
                } else {
                    insertList.add(record);
                }
            }

            // 批量插入与更新
            if (!insertList.isEmpty()) {
                log.info("任务ID：{} 开始批量插入 {} 条数据", taskId, insertList.size());
                batchInsert(targetConn, task, insertList, fieldMapping);
                log.info("任务ID：{} 批量插入完成", taskId);
            }
            if (!updateList.isEmpty()) {
                log.info("任务ID：{} 开始批量更新 {} 条数据", taskId, updateList.size());
                batchUpdate(targetConn, task, updateList, fieldMapping);
                log.info("任务ID：{} 批量更新完成", taskId);
            }

            // 更新同步时间
            String maxSyncTime = SDF_DATETIME.format(maxTime);
            recordSyncTime(task, maxSyncTime);
        }
    }

    /**
     * 执行全量数据查询，从源数据库读取所有数据。
     * 该方法不涉及事务管理，仅用于数据读取操作。
     *
     * @param task 同步任务配置对象，包含源库信息及字段映射规则
     * @param conn 数据库连接对象，由调用方负责连接的创建与关闭
     * @return 查询结果，以Map列表形式返回，每个Map代表一行数据
     * @throws SQLException 如果数据库查询过程中发生异常
     */
    private List<Map<String, Object>> loadAllSourceData(SyncTask task, Connection conn) throws SQLException {
        // 只查询字段映射中定义的源字段
        String fields = String.join(", ", task.getFieldMappingMap().keySet());
        String sql = "SELECT " + fields + " FROM " + task.getSourceTable();

        log.info("任务ID：{} 执行全量查询SQL: {}", task.getId(), sql);

        try (PreparedStatement stmt = conn.prepareStatement(sql);
             ResultSet rs = stmt.executeQuery()) {
            return convertToList(rs);
        } catch (SQLException e) {
            log.error("任务ID：{} 全量查询失败 - SQL: {}, 错误: {}",
                    task.getId(), sql, e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 执行增量数据查询，从源数据库读取指定时间点之后的数据。
     * 该方法不涉及事务管理，仅用于数据读取操作。
     *
     * @param task 同步任务配置对象，包含源库信息、时间字段及同步状态
     * @param conn 数据库连接对象，由调用方负责连接的创建与关闭
     * @return 查询结果，以Map列表形式返回，每个Map代表一行数据
     * @throws SQLException 如果数据库查询过程中发生异常
     */
    private List<Map<String, Object>> loadNewSourceDataSince(SyncTask task, Connection conn) throws SQLException {
        // 只查询字段映射中定义的源字段
        String fields = String.join(", ", task.getFieldMappingMap().keySet());

        // 根据数据库类型构建不同的时间比较条件
        String timeCondition = buildTimeComparisonCondition(task);
        String sql = "SELECT " + fields + " FROM " + task.getSourceTable() + " WHERE " + timeCondition;

        log.info("任务ID：{} 执行增量查询SQL: {}", task.getId(), sql);

        try (PreparedStatement stmt = conn.prepareStatement(sql)) {
            String dateString = SDF_DATETIME.format(task.getLastSyncTime());

            // 设置时间参数
            stmt.setString(1, dateString);
            log.debug("任务ID：{} 查询参数 - 时间条件: {}", task.getId(), dateString);

            try (ResultSet rs = stmt.executeQuery()) {
                return convertToList(rs);
            }
        } catch (SQLException e) {
            log.error("任务ID：{} 增量查询失败 - SQL: {}, 错误: {}",
                    task.getId(), sql, e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 根据数据库类型构建时间比较条件
     */
    private String buildTimeComparisonCondition(SyncTask task) {
        String timeField = task.getSourceTimeField();
        String dbType = task.getSourceDbType();

        // 根据不同数据库类型返回不同的时间字段转换表达式
        switch (dbType.toLowerCase()) {
            case "postgresql":
                // Postgresql: 使用to_char转换为秒级精度
                return "to_char(" + timeField + ", 'YYYY-MM-DD HH24:MI:SS') > ?";
            case "oracle":
                // Oracle: 使用TO_CHAR转换为秒级精度
                return "TO_CHAR(" + timeField + ", 'YYYY-MM-DD HH24:MI:SS') > ?";
            case "mysql":
                // MySQL: 使用DATE_FORMAT转换为秒级精度
                return "DATE_FORMAT(" + timeField + ", '%Y-%m-%d %H:%i:%s') > ?";
            case "sqlserver":
                // SQL Server: 使用CONVERT转换为秒级精度
                return "CONVERT(VARCHAR, " + timeField + ", 120) > ?";
            default:
                // 其他数据库直接比较(如SQLite)
                return timeField + " > ?";
        }
    }

    /**
     * 批量插入数据到目标表
     *
     * @param conn         数据库连接
     * @param task         同步任务
     * @param dataList     待插入数据列表
     * @param fieldMapping 字段映射关系
     * @throws SQLException 数据库异常
     */
    private void batchInsert(Connection conn, SyncTask task, List<Map<String, Object>> dataList,
                             Map<String, String> fieldMapping) throws SQLException {
        int batchSize = task.getBatchSize() != null ? task.getBatchSize() : 50;
        String columns = String.join(", ", fieldMapping.values());
        String placeholders = String.join(", ", Collections.nCopies(fieldMapping.size(), "?"));
        String sql = String.format("INSERT INTO %s (%s) VALUES (%s)",
                task.getTargetTable(), columns, placeholders);

        int totalCount = dataList.size();
        log.info("任务ID：{} 开始批量插入 {} 条数据，批次大小 {} 条",
                task.getId(), totalCount, batchSize);

        try (PreparedStatement stmt = conn.prepareStatement(sql)) {
            long startTime = System.currentTimeMillis();
            int processedCount = 0;
            int totalSuccess = 0;
            int totalFailed = 0;

            for (Map<String, Object> data : dataList) {
                // 设置参数
                int index = 1;
                for (String targetField : fieldMapping.values()) {
                    String sourceField = getKeyByValue(fieldMapping, targetField);
                    stmt.setObject(index++, data.get(sourceField));
                }
                stmt.addBatch();
                processedCount++;

                // 达到批次大小时执行
                if (processedCount % batchSize == 0 || processedCount == totalCount) {
                    int[] results = stmt.executeBatch();
                    stmt.clearBatch();

                    // 统计本批次结果
                    BatchResult batchResult = analyzeBatchResults(results);
                    totalSuccess += batchResult.successCount;
                    totalFailed += batchResult.failedCount;

                    log.info("任务ID：{} 插入进度: {}/{} ({}%)，本批次成功: {}条，失败: {}条",
                            task.getId(),
                            processedCount,
                            totalCount,
                            (processedCount * 100 / totalCount),
                            batchResult.successCount,
                            batchResult.failedCount);
                }
            }

            long endTime = System.currentTimeMillis();
            log.info("任务ID：{} 批量插入完成，共处理 {} 条，成功 {} 条，失败 {} 条，总耗时 {} 毫秒",
                    task.getId(), totalCount, totalSuccess, totalFailed, (endTime - startTime));

            if (totalFailed > 0) {
                log.warn("任务ID：{} 存在插入失败记录，成功/失败/总数: {}/{}/{}",
                        task.getId(), totalSuccess, totalFailed, totalCount);
            }
        } catch (SQLException e) {
            log.error("任务ID：{} 批量插入过程中发生异常 - {}", task.getId(), e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 批量更新目标表数据
     *
     * @param conn         数据库连接
     * @param task         同步任务
     * @param dataList     待更新数据列表
     * @param fieldMapping 字段映射关系
     * @throws SQLException 数据库异常
     */
    private void batchUpdate(Connection conn, SyncTask task,
                             List<Map<String, Object>> dataList,
                             Map<String, String> fieldMapping) throws SQLException {
        int batchSize = task.getBatchSize() != null ? task.getBatchSize() : 50;
        StringBuilder setClause = new StringBuilder();
        for (String targetField : fieldMapping.values()) {
            if (setClause.length() > 0) setClause.append(", ");
            setClause.append(targetField).append(" = ?");
        }

        String whereClause = task.getTargetIdField() + " = ?";
        String sql = String.format("UPDATE %s SET %s WHERE %s",
                task.getTargetTable(), setClause, whereClause);

        int totalCount = dataList.size();
        log.info("任务ID：{} 开始批量更新 {} 条数据，批次大小 {} 条",
                task.getId(), totalCount, batchSize);

        try (PreparedStatement stmt = conn.prepareStatement(sql)) {
            long startTime = System.currentTimeMillis();
            int processedCount = 0;
            int totalSuccess = 0;
            int totalFailed = 0;

            for (Map<String, Object> data : dataList) {
                // 设置更新字段参数
                int index = 1;
                for (String targetField : fieldMapping.values()) {
                    String sourceField = getKeyByValue(fieldMapping, targetField);
                    stmt.setObject(index++, data.get(sourceField));
                }

                // 设置WHERE条件参数
                String idField = task.getSourceIdField();
                stmt.setObject(index, data.get(idField));

                stmt.addBatch();
                processedCount++;

                // 达到批次大小时执行
                if (processedCount % batchSize == 0 || processedCount == totalCount) {
                    int[] results = stmt.executeBatch();
                    stmt.clearBatch();

                    // 统计本批次结果
                    BatchResult batchResult = analyzeBatchResults(results);
                    totalSuccess += batchResult.successCount;
                    totalFailed += batchResult.failedCount;

                    log.info("任务ID：{} 更新进度: {}/{} ({}%)，本批次成功: {}条，失败: {}条",
                            task.getId(),
                            processedCount,
                            totalCount,
                            (processedCount * 100 / totalCount),
                            batchResult.successCount,
                            batchResult.failedCount);
                }
            }

            long endTime = System.currentTimeMillis();
            log.info("任务ID：{} 批量更新完成，共处理 {} 条，成功 {} 条，失败 {} 条，总耗时 {} 毫秒",
                    task.getId(), totalCount, totalSuccess, totalFailed, (endTime - startTime));

            if (totalFailed > 0) {
                log.warn("任务ID：{} 存在更新失败记录，成功/失败/总数: {}/{}/{}",
                        task.getId(), totalSuccess, totalFailed, totalCount);
            }
        } catch (SQLException e) {
            log.error("任务ID：{} 批量更新过程中发生异常 - {}", task.getId(), e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 记录本次同步的最后同步时间到数据库中。
     *
     * @param task 同步任务对象，包含任务ID和最后同步时间
     */
    public void recordSyncTime(SyncTask task, String maxSyncTime) {
        Date lastSyncTime = task.getLastSyncTime();
        Integer taskId = task.getId();
        if (lastSyncTime == null) {
            syncConfigMapper.saveLastSyncTime(taskId, maxSyncTime);
        } else {
            syncConfigMapper.updateLastSyncTime(taskId, maxSyncTime);
        }
        log.info("任务ID：{} 记录最后同步时间成功，时间：{}", taskId, maxSyncTime);
    }

    /**
     * 检查目标数据库中是否已存在指定ID的数据。
     *
     * @param conn    数据库连接
     * @param task    同步任务配置
     * @param idValue 主键值
     * @return true 如果存在相同主键的数据，false 否则
     * @throws SQLException 数据库访问异常
     */
    private boolean checkIfExists(Connection conn, SyncTask task, Object idValue) throws SQLException {
        String sql = "SELECT 1 FROM " + task.getTargetTable() +
                " WHERE " + task.getTargetIdField() + " = ?";

        log.debug("任务ID：{} 执行存在性检查SQL: {}, 参数: {}",
                task.getId(), sql, idValue);

        try (PreparedStatement stmt = conn.prepareStatement(sql)) {
            stmt.setObject(1, idValue);
            try (ResultSet rs = stmt.executeQuery()) {
                return rs.next();
            }
        } catch (SQLException e) {
            log.error("任务ID：{} 存在性检查失败 - SQL: {}, 参数: {}, 错误: {}",
                    task.getId(), sql, idValue, e.getMessage(), e);
            throw e;
        }
    }

    /**
     * 将 ResultSet 转换为 Map 的 List 集合。
     * 每一行数据被转换为一个 Map，其中键为列名，值为对应的数据内容。
     *
     * @param rs 数据库查询结果集对象
     * @return 包含结果数据的 Map 列表，每个 Map 表示一行数据
     * @throws SQLException 如果在读取结果集时发生数据库异常
     */
    private List<Map<String, Object>> convertToList(ResultSet rs) throws SQLException {
        List<Map<String, Object>> list = new ArrayList<>();
        ResultSetMetaData metaData = rs.getMetaData();
        while (rs.next()) {
            Map<String, Object> row = new HashMap<>();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                row.put(metaData.getColumnName(i), rs.getObject(i));
            }
            list.add(row);
        }
        return list;
    }

    /**
     * 分析批处理结果
     *
     * @param results 批处理结果数组
     * @return 包含成功和失败计数的结果对象
     */
    private BatchResult analyzeBatchResults(int[] results) {
        BatchResult result = new BatchResult();
        for (int r : results) {
            if (r >= 0 || r == Statement.SUCCESS_NO_INFO) {
                result.successCount++;
            } else {
                result.failedCount++;
            }
        }
        return result;
    }

    /**
     * 批处理结果统计类
     */
    private static class BatchResult {
        int successCount = 0;
        int failedCount = 0;
    }
}
